package Flink.transformation;

import Flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggregationTrans {
    // Todo 简单聚合（sum/min/max/minBy/maxBy）
    /*
    *   ⚫ sum()：在输入流上，对指定的字段做叠加求和的操作。
        ⚫ min()：在输入流上，对指定的字段求最小值。
        ⚫ max()：在输入流上，对指定的字段求最大值。
        ⚫ minBy()：与 min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而 minBy()则会返回包含字段最小值的整条数据。
        ⚫ maxBy()：与 max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。
    *
    * */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_1", 2L, 2),
                new WaterSensor("sensor_2", 2L, 2),
                new WaterSensor("sensor_3", 3L, 3)
        );

        stream.keyBy(data -> data.id).sum("vc").print();
        /*
        * 一个聚合算子，会为每一个 key 保存一个聚合的值，在 Flink 中我们把它叫作“状态”（state）。
        * 所以每当有一个新的数据输入，算子就会更新保存的聚合结果，
        * 并发送一个带有更新后聚合值的事件到下游算子。
        * 对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子，应该只用在含有有限个key的数据流上。
        * */
        env.execute();
    }
}
